fix: fall back to Spark for NOT IN over an ordered streamed side by dwsmith1983 · Pull Request #6810 · apache/datafusion-comet

dwsmith1983 · 2026-10-09T05:58:55Z

Which issue does this PR close?

Part of #6787.

Rationale for this change

Spark plans NOT IN as a null-aware anti join that streams the left side, and the join reports that side's ordering. When the streamed side arrives sorted, for example from a sort-merge join, a window, or a cached table sorted within its partitions, Spark drops a sort above the join, and a sort aggregate above it skips its own sort.

Comet runs a null-aware anti join unswapped in CollectLeft mode, so DataFusion builds on Spark's streamed side. With a single integer key DataFusion collects the key range even without a dynamic filter, and picks its array map when the range is small or the keys are dense. The array map keeps the build batches in order. Otherwise it builds a hash map, which concatenates the build batches in reverse (datafusion-physical-plan 55.2, hash_join/exec.rs:2344 and 2365), so with sparse keys and more than one build batch the unmatched rows come out in reverse batch order. sortWithinPartitions then returns unsorted partitions, and a Spark sort aggregate above the join splits groups (223 groups instead of 90 in the new test). The upstream issue is apache/datafusion#26226.

This happens in fully native plans on main too: a native window over a native sort, read from parquet with default settings, feeds the native join and comes back out of order. The cached table case needs spark.comet.convert.inMemoryCache.enabled, but the window case does not.

An earlier version of this PR sorted the native join output instead. Under the fair pool that cost about 4.2x the memory: 39 MB on main against 165 MB with the sort, at local[1] under fair_unified with a 2M-row sorted partition. Falling back for this one shape uses less memory than a native sort and is simpler.

What changes are included in this PR?

CometHashJoin falls back to Spark when the join is a null-aware anti join and Spark reports an output ordering for it. A comment above the guard links HashJoinExec loses build-side input order across batches on the hash map path datafusion#26226, so it can go once a DataFusion release with that fix reaches Comet. A NOT IN whose streamed side has no ordering still runs natively, and so do other hash joins over an ordered side, such as a semi join.
The compatibility guide gets a hash join note, and the BroadcastHashJoinExec row in the operator table lists the fallback.
The ExistenceJoin docs and the spark.comet.exec.existenceJoin.enabled description no longer list NOT IN among the existence join fallbacks. NOT IN combined with OR plans as an existence BroadcastNestedLoopJoinExec, which is what falls back.

This PR no longer adds an output_ordering field to the HashJoin proto and changes no native code. Outer hash joins over a streamed side sorted outside the native plan are handled in #6835.

Behavior change against branch-1.1: there, every NOT IN over an ordered streamed side (sort-merge join, sort, window or sorted cache output) runs natively and can return rows out of order. With this PR that join runs in Spark until a DataFusion release with apache/datafusion#26226 reaches Comet, which gives up some speed for correct order. branch-1.1 has the same code path, so this is a backport candidate.

How are these changes tested?

New CometJoinSuite tests with sparse keys and a small batch size, so the build spans many batches:

A NOT IN over a sorted cached streamed side falls back with the new reason, with AQE on and off. Rows match Spark, no sort runs above the join, and every partition is in order.
A NOT IN below an outer hash join over the same cached side, grouped with a Spark sort aggregate, returns Spark's groups and keeps every partition in order.
A NOT IN over a native window and sort read from parquet, with default settings apart from the batch size, falls back and keeps every partition in order, with AQE on and off. The fallback reason is checked with AQE off: with AQE on, the join is planned again after the shuffle stage, and only the initial plan carries the reason.
A LEFT SEMI join over that same window side still runs as a native CometBroadcastHashJoinExec, and its partitions stay in order.
A NOT IN over an unordered scan still runs as a native CometBroadcastHashJoinExec.

The first three fail on main (Spark 3.5 and 4.1 checked) and pass with the change. The cached NOT IN check run natively on main with dense keys (_1 = i, Spark 3.5) keeps its order, which matches the array map path above. CometJoinSuite passes on Spark 3.4, 3.5, 4.0 and 4.1, and the strict warnings build passes.

… is not kept Spark reports a hash join's streamed-side ordering and may drop a sort above the join on that basis. DataFusion keeps unmatched probe rows of a right join in order only when it sees the probe input sorted, which it does not when that order comes from outside the native plan. The new HashJoin output_ordering field carries the ordering, and the native planner adds a sort above the join when the join does not already produce it.

For LeftOuter with BuildRight and RightOuter with BuildLeft, Spark reports the streamed side's ordering, and DataFusion runs the join as a right join that may emit unmatched probe rows after the matched ones. The serde now sends that ordering so the native planner can restore it, and falls back to Spark when the ordering cannot be sorted natively.

The ordering the merge join produces reaches the hash join through the swapped right join and its projection, so the planner keeps it without sorting again.

…ordering tests A null-aware anti join runs unswapped, so DataFusion builds on Spark's streamed side and concatenates its batches in reverse; it now sends its ordering too and gets the same native sort. The tests cover multi-column and multi-batch orderings, the expression fallback, a natively sorted probe, join metrics with a sort on top, and the join class per hint. The compatibility guide describes how hash joins keep the order.

The null-aware anti join test now checks that its build spans several batches, a metric check that held for any batch size is gone, and a test comment claims only what the test checks. The guides say a NOT IN join is sorted too, and that only a NOT IN under OR, planned as an existence nested loop join, falls back to Spark.

…rder

…atches DataFusion builds a NOT IN join on Spark's streamed side and, on its hash path, concatenates those batches in reverse, so the planner sorts the output. The test feeds two sparse-key batches and checks the order.

A plain NOT IN runs as a native null-aware anti join; only one combined with another predicate through OR becomes an existence join that falls back to Spark.

…rder

dwsmith1983 · 2026-10-09T07:50:39Z

@andygrove could you take a look when you get a chance? I'd also suggest backport-1.1: branch-1.1 has the same code path, where the hash join execs copy Spark's streamed ordering and the native join keeps no probe order for these shapes. Its tests would turn on spark.comet.sparkToColumnar.enabled, since 1.1 has no spark.comet.convert.inMemoryCache.enabled. #6785 should merge first, as the description says.

sunchao

Reviewed the full seven-file diff from 00a4b422f0ed6560ef76dce004b94c3697613108 to 8482fc93b5b4a51fd506c44cc0adf731279504bf. The PR is not a draft. No introduced P1/P2 issues found within this review. Existing discussion contains no substantiated unresolved P1/P2 concern.

Summary

Prior state and problem: Spark can remove a sort because a hash join reports its streamed input’s ordering. Native outer joins could move unmatched rows, while null-aware anti joins could reverse build-batch order, violating that promise.
Design approach: Serialize the reported ordering for the affected join shapes and restore it with SortExec when DataFusion’s equivalence properties cannot guarantee it.
Correctness: Checked Spark’s HashJoin.outputOrdering, ShuffledHashJoinExec exceptions, RemoveRedundantSorts, and ordering tests against DataFusion’s execution paths. The sort follows the join’s output-restoring projection, so bound column indices remain valid. Existing sort serialization and normalization preserve direction, null placement, and floating-point handling. No P1/P2 issue identified.
Compatibility analysis: Checked the relevant sources for Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. The affected ordering contract is consistent across them. branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1 has the original behavior, making this an intended correctness change and a backport candidate. The protobuf field is additive and defaults to empty.
Key design decisions: Restricting serialization to the two affected outer shapes and null-aware anti joins limits additional work. Unsupported ordering types or expressions fall back to Spark through existing compatibility checks.
Implementation sketch: CometHashJoin.doConvert binds ordering expressions to join.output. The planner handles join swapping, checks ordering satisfaction, and conditionally wraps the result. additional_native_plans retains join metrics without double-counting output rows.
Performance: Unordered joins and joins whose native ordering already satisfies the requirement avoid the extra sort. Restoring missing order adds buffering, sorting, and two memory consumers, reducing other consumers’ shares under the fair pool. A bounded execution probe successfully exercised spilling. No throughput improvement or regression was inferred without measurements.
Design: Reusing DataFusion’s spillable sorter keeps the correction local and follows the existing sort-aggregate approach. The implementation is straightforward to reason about.
Abstraction & complexity: sort_unless_ordered cleanly isolates the ordering check and wrapper construction. Reusing existing serialization, normalization, and metric aggregation avoids introducing another sorting abstraction.
Behavioral changes worth calling out: Affected joins now restore their advertised order across unmatched rows and batch boundaries. Unsupported orderings can newly trigger fallback. The NOT IN documentation correction changes neither configuration defaults nor execution routing.
Suggested improvements: No additional code change meeting the P1/P2 reporting bar was identified.

Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-memory-pr. Read AGENTS.md, applicable contributor guidance, the supplied discussion snapshot, and current GitHub reviews/comments. No Copilot feedback was used.

Exact-head CI: Label pull requests passed. Comet CI, Check PR Title, and CodeQL report action_required. There is no exact-head build/test verdict yet.

Validation: A disposable standalone probe using DataFusion 55.1.0 and Arrow 59.3.0 reproduced both outer-join reorderings and the sparse-key null-aware anti-join reordering, then verified that sorting restores order without changing rows. Both outer shapes avoided an extra sort with natively sorted probes. A 100,000-row case returned correctly ordered output under a 256 KiB pool with 14 spills. git diff --check passed.

Validation limits: The focused exact-head native test build failed during dependency compilation with No space left on device. Its artifacts were removed. Comet planner tests, CometJoinSuite, Spark SQL suites, and the actual Comet fair-memory-pool path were not independently executed. The standalone probe validates the DataFusion mechanism, not the complete JVM/protobuf/native integration. Author-reported suite passes remain unverified. Project code is unchanged.

comphead

I checked the DataFusion 55.1.0 ordering paths and Spark's outputOrdering for the affected shapes and found no correctness problem. The inline comments cover sharing the new sort helper with the sort aggregate path and folding one repeated test run into the existing matrix.

comphead · 2026-10-09T16:11:09Z


+    /// Returns a sort of `plan` by the SortOrder exprs in `ordering`, or `None` when `ordering`
+    /// is empty or the plan's equivalence properties already satisfy it.
+    fn sort_unless_ordered(


This is the same check and wrap that the sort aggregate already does for ordered_by_grouping_keys (the ordering_satisfy block in the HashAgg arm, around line 1712). Could sort_unless_ordered take the PhysicalSortExprs, or a LexOrdering, instead of Spark Exprs, with the join mapping create_sort_expr at its call site? Then the aggregate arm could call it too, and the planner would keep one place that decides when an output sort is needed.

Could sort_unless_ordered take the PhysicalSortExprs, or a LexOrdering, instead of Spark Exprs, with the join mapping create_sort_expr at its call site?

Done. The helper takes the physical sort expressions, the hash join maps create_sort_expr where it calls it, and the sort aggregate arm now goes through the same helper.

comphead · 2026-10-09T16:11:09Z

+          val (leftOuter, buildSide) =
+            checkJoinKeepsOrder(query("left_outer"), keyOrder(IntegerType, Ascending))
+          assert(buildSide == BuildRight)
+          assert(leftOuter.nativeOp.getHashJoin.getOutputOrderingCount > 0)


The left_outer half of this test runs the same tables, cache, query and confs as the matrix case shuffle_hash left_outer ... condition=false, AQE=false, batchSize=8192 above, so it repeats that run. Could the getOutputOrderingCount > 0 assertion move into the matrix loop instead? That would check the field for every affected shape, including broadcast and right_outer, and this test could keep only the inner check.

Could the getOutputOrderingCount > 0 assertion move into the matrix loop instead?

Moved. Every matrix case now asserts that the join sends its ordering, and the separate test keeps only the inner check.

…t aggregate sort_unless_ordered now takes physical sort expressions. The hash join builds them at its call site, and the sort aggregate arm calls the same helper instead of its own ordering_satisfy check and SortExec wrap.

The streamed order matrix and the null-aware anti join test assert that the native op carries an output ordering. The separate test keeps only the inner join check, since its outer half repeated a matrix case.

…rder

With runtime filters on, a probe-side filter or projection becomes CometFilterExec or CometProjectionExec before the join is built. Check that both keep the sort's ordering, so an outer hash join that reports its streamed ordering still adds no sort.

…rder

andygrove

Thanks for offering to narrow this to NOT IN. At dc5e8fc7de it still sorts the two outer shapes, which #6835 now covers without a sort, so I'll look again once the NOT IN version is up. I'm marking this as request changes so an approval doesn't merge this head before then. The docs need the same trim, and the ShuffledHashJoinExec row can go back to empty since only broadcast joins are null-aware. A few things for that version are inline.

andygrove · 2026-10-12T12:49:07Z

+                    .map(|expr| self.create_sort_expr(expr, Arc::clone(&join_schema)))
+                    .collect::<Result<Vec<_>, _>>()?;
+                let native_plan =
+                    match Self::sort_unless_ordered(Arc::clone(&join_root), output_ordering)? {


For a NOT IN, the build DataFusion holds here is Spark's streamed side, the whole partition, and it can't spill. SortExec::execute registers ExternalSorter[0] and ExternalSorterMerge[0] before that build starts, and fair_unified gives every registered consumer the same share. So the build's limit goes from pool/N to pool/(N+2), a third of the pool when the join was the only consumer. That only binds when a task has more consumers than the executor runs tasks at once, but on one- or two-core executors a NOT IN that fit on main can now fail in HashJoinInput. The sort's first batch also has to reserve 10 MB for the merge plus about twice the batch before it has anything to spill. Could you run a NOT IN over a sorted streamed side sized near the limit at local[1], on main and on this branch with spark.comet.debug.memory=true, so we know what this costs? The reorder itself comes from collect_left_input concatenating the build batches in reverse. Inserting them with descending offsets and concatenating forward would keep both the chain order and the input order, so an upstream fix there would let this sort go away later.

Inserting them with descending offsets and concatenating forward would keep both the chain order and the input order, so an upstream fix there would let this sort go away later.

Filed apache/datafusion#26226 for that. #6810 keeps the sort until a DataFusion release with the fix reaches Comet, and the sort comes out then. The local[1] memory comparison will come with the NOT IN version.

Could you run a NOT IN over a sorted streamed side sized near the limit at local[1], on main and on this branch with spark.comet.debug.memory=true, so we know what this costs?

It costs more than the share alone. At local[1] under fair_unified, a NOT IN over a 2M-row sorted partition (1.8M rows survive) needs a 39 MB pool on main and 165 MB with the sort:

Below 116 MB the build fails in HashJoinInput. ExternalSorter[0] and ExternalSorterMerge[0] register first, so the build gets a third of the pool.

From 116 to 163 MB the build fits, but the sort fails with "Not enough memory to continue external sort". The join hands it all 1.8M rows as one 28.8 MB batch, and ExternalSorter has to reserve twice that inside its own third with nothing to spill.

So the sort is gone. This version falls back to Spark for a NOT IN whose streamed side reports an ordering, and the guard's comment points at apache/datafusion#26226 so it can come out once that fix reaches Comet.

andygrove · 2026-10-12T12:49:07Z

+                let native_plan =
+                    match Self::sort_unless_ordered(Arc::clone(&join_root), output_ordering)? {
+                        Some(sort) => {
+                            additional_native_plans.push(join_root);


The sort's metrics fold into the join node here, but CometMetricNode.joinMetrics has no spill_count, spilled_bytes or spilled_rows. A spill in this sort is dropped from the SQL metrics, and it never reaches the task's diskBytesSpilled either, because reportSpillMetrics only sums metrics the node declares. output_batches will also count the sort's batches on top of the join's. Could the hash join execs add the spill metrics when getOutputOrderingCount > 0, the way they add joinDynamicFilterMetrics?

Could the hash join execs add the spill metrics when getOutputOrderingCount > 0, the way they add joinDynamicFilterMetrics?

With the sort gone there is no second plan under the join, so the node's metrics are the same as on main.

andygrove · 2026-10-12T12:49:07Z

+    }
+  }
+
+  test("hash join falls back when its streamed ordering has an unsupported sort type") {


When this narrows to NOT IN, could these cases move to the NOT IN shape rather than go with the outer ones? Both fallback tests, the descending nulls-last order, NaN with -0.0, and the two-key order are the only coverage of the new serde path. For nulls last, sorting by a nullable column other than the key is simpler, because NOT IN drops null keys whenever the subquery isn't empty. On the planner side, a NOT IN whose streamed side is a native sort should still get the SortExec, since DataFusion's LeftAnti reports maintains_input_order as [false, false]. That's the case where the input looks sorted and the output isn't, and no test covers it yet.

When this narrows to NOT IN, could these cases move to the NOT IN shape rather than go with the outer ones?

The descending, NaN and two-key cases tested how the sort's ordering was serialized, and that code is gone with the sort. The NOT IN tests now cover:

a sorted cache;

a native window over a native sort read from parquet with default settings, which is the case where the input looks sorted and the output isn't (both with AQE on and off);

a LEFT SEMI join over that same window side, which stays native;

a NOT IN over an unordered scan, which stays native.

Comet runs a null-aware anti join unswapped in CollectLeft mode, so DataFusion builds on Spark's streamed side. On its hash map path the build batches are concatenated in reverse, and the unmatched rows come out in reverse batch order. Spark has already dropped any sort above the join because the join reports the streamed side's ordering, so sortWithinPartitions returns unsorted partitions and a sort aggregate above the join splits groups. Sorting the native output instead needed about four times the memory pool under fair_unified, so fall back to Spark when a null-aware anti join's streamed side reports an ordering, until the DataFusion fix (apache/datafusion#26226) reaches Comet. This drops the HashJoin output_ordering field and the planner sort from this branch.

…rder

dwsmith1983 · 2026-10-12T15:59:41Z

I'll look again once the NOT IN version is up.

It's up. It no longer sorts: the memory run (details in the thread on planner.rs) showed the sort needs about four times the pool, so a NOT IN whose streamed side reports an ordering now falls back to Spark. There is no HashJoin field and no native change any more. The docs cover NOT IN only, and the ShuffledHashJoinExec row is empty again.

…rder

dwsmith1983 added 6 commits October 9, 2026 09:09

test: a native sort-merge join below a hash join adds no extra sort

7f67a6d

The ordering the merge join produces reaches the hash join through the swapped right join and its projection, so the planner keeps it without sorting again.

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

3f4aabd

…rder

github-actions Bot added the bug Something isn't working label Oct 9, 2026

dwsmith1983 added 3 commits October 9, 2026 14:26

docs: say only a NOT IN combined with OR skips the native existence join

7336bac

A plain NOT IN runs as a native null-aware anti join; only one combined with another predicate through OR becomes an existence join that falls back to Spark.

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

8482fc9

…rder

sunchao reviewed Oct 9, 2026

View reviewed changes

comphead reviewed Oct 9, 2026

View reviewed changes

dwsmith1983 added 9 commits October 10, 2026 01:18

test: check the hash join ordering field for every shape that sends one

1c9b7e9

The streamed order matrix and the null-aware anti join test assert that the native op carries an output ordering. The separate test keeps only the inner join check, since its outer half repeated a matrix case.

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

3e9adee

…rder

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

fc2340a

…rder

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

0e20d4e

…rder

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

6e8af20

…rder

Merge branch 'main' into fix/hash-join-probe-order

44b679e

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

dc5e8fc

…rder

andygrove mentioned this pull request Oct 11, 2026

fix: keep a streamed side sorted outside the native plan in order through outer hash joins #6835

Open

andygrove requested changes Oct 12, 2026

View reviewed changes

This was referenced Oct 12, 2026

HashJoinExec loses build-side input order across batches on the hash map path apache/datafusion#26226

Open

fix: keep build-side input order across batches in hash join apache/datafusion#26228

Open

Merge commit 'cf6673808' into fix/hash-join-probe-order

0946593

andygrove mentioned this pull request Oct 12, 2026

HashJoinExec: declare build-side order for LeftSemi, LeftAnti and LeftMark joins apache/datafusion#26235

Open

dwsmith1983 added 2 commits October 12, 2026 22:45

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

f73829f

…rder

dwsmith1983 changed the title ~~fix: keep a native hash join's streamed order when Spark drops the sort above it~~ fix: fall back to Spark for NOT IN over an ordered streamed side Oct 12, 2026

andygrove mentioned this pull request Oct 12, 2026

Native broadcast nested loop joins report the streamed side's order without keeping it #6898

Open

dwsmith1983 and others added 5 commits October 12, 2026 23:56

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

dfc848d

…rder

Merge branch 'main' into fix/hash-join-probe-order

732dd85

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

58b827d

…rder

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

7249b46

…rder

Merge remote-tracking branch 'origin/main' into fix/hash-join-probe-o…

08bc29b

…rder

Conversation

dwsmith1983 commented Oct 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Uh oh!

dwsmith1983 commented Oct 9, 2026

Uh oh!

sunchao left a comment

Choose a reason for hiding this comment

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andygrove left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dwsmith1983 commented Oct 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

dwsmith1983 commented Oct 9, 2026 •

edited

Loading